
package com.wmx.active_mq_1;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * 非持久化 topic(主题) 消息消费者
 */
@SuppressWarnings("all")
public class NoPersistenceTopicReceiver {
    public static void main(String[] args) {
        Connection connection = null;
        Session session = null;
        MessageConsumer messageConsumer = null;
        try {
            String brokerURL = "tcp://127.0.0.1:61616";//ActiveMQ 中间件连接地址
            ConnectionFactory mqConnectionFactory = new ActiveMQConnectionFactory(brokerURL);
            //如果 ActiveMQ 连不上，则抛异常：java.net.ConnectException: Connection refused: connect
            connection = mqConnectionFactory.createConnection();//通过连接工厂获取连接 javax.jms.Connection
            connection.start();//启动连接，同理还有 stop、close

            //创建会话 session。开启事务，消息确认模式为自动确认
            session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

            //创建消息主题(topic)，主题名称与生产者设置的保持一致。Queue 、Topic 都是 Destination 的字接口
            Destination destination = session.createTopic("topic-app");

            messageConsumer = session.createConsumer(destination);//根据目的地创建消息消费者
            Message message = messageConsumer.receive();//receive方法会导致当前线程阻塞，直到接收到消息

            while (message != null) {
                if (message instanceof TextMessage) {
                    TextMessage textMessage = (TextMessage) message;
                    System.out.println("收到消息：" + textMessage.getText());
                    session.commit();//确认消息。
                    //接收消息时设置超时时间，单位为毫秒。如果为0，则等同于 receive()一致阻塞。
                    //如果超过超时时间，仍然未接收到消息，则返回 null。while 会推出
                    message = messageConsumer.receive(3000);
                }
            }
        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            try {
                if (session != null) {
                    session.close();//关闭会话
                }
                if (connection != null) {
                    connection.close();//关闭连接
                }
                if (messageConsumer != null) {
                    messageConsumer.close();//关闭消费者
                }
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}
